package com.atguigu.edu.app.dws;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.bean.ProvinceTransactionOrderInfo;
import com.atguigu.edu.util.KafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.time.Duration;

public class DwsProvinceTransactionWindow {
    public static void main(String[] args) {
        //todo 1 初始化环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //todo 2 设置状态后端
        /*env.enableCheckpointing(5 * 1000L, CheckpointingMode.AT_LEAST_ONCE);
        env.getCheckpointConfig().setCheckpointTimeout(3 * 60 *1000L);
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall/ck");
        System.setProperty("HADOOP_USER_NAME","atguigu");
        env.setStateBackend(new HashMapStateBackend());*/

        //todo 3 从kafka获取主题数据
        String topicName = "dwd_trade_order_detail";
        String groupId = "dws_province_transaction_window";
        DataStreamSource<String> dataStreamSource = env.addSource(KafkaUtil.getKafkaConsumer(topicName, groupId));


        //todo 4 转换数据类型
        SingleOutputStreamOperator<ProvinceTransactionOrderInfo> mapStream = dataStreamSource.map(new MapFunction<String, ProvinceTransactionOrderInfo>() {
            @Override
            public ProvinceTransactionOrderInfo map(String value) throws Exception {
                JSONObject jsonObject = JSONObject.parseObject(value);
                ProvinceTransactionOrderInfo orderInfo = new ProvinceTransactionOrderInfo();
                orderInfo.setProvinceId(jsonObject.getString("province_id"));
                orderInfo.setUserId(jsonObject.getString("user_id"));
                orderInfo.setOriginalAmount(jsonObject.getDouble("origin_amount"));
                orderInfo.setCouponAmount(jsonObject.getDouble("coupon_reduce"));
                orderInfo.setOriginalAmount(jsonObject.getDouble("final_amount"));
                orderInfo.setTs(jsonObject.getLong("ts"));
                return orderInfo;
            }
        });

        //todo 5 设置水位线
        SingleOutputStreamOperator<ProvinceTransactionOrderInfo> watermaekStream = mapStream.assignTimestampsAndWatermarks(WatermarkStrategy
                .<ProvinceTransactionOrderInfo>forBoundedOutOfOrderness(Duration.ofSeconds(2L))
                .withTimestampAssigner(new SerializableTimestampAssigner<ProvinceTransactionOrderInfo>() {
                    @Override
                    public long extractTimestamp(ProvinceTransactionOrderInfo element, long recordTimestamp) {
                        return element.getTs() * 1000L;
                    }
                }));



    }
